package com.etc.sparkstrea

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object WindowHotWord  {
  def main(args: Array[String]): Unit = {
      val conf = new SparkConf().setAppName("WindowHotWord").setMaster("local[*]")
      val sc = new StreamingContext(conf,Seconds(1))
      val line = sc.socketTextStream("master",9999)
    val unit = line.map(a => (a.split(" ")(1), 1)).reduceByKeyAndWindow(
      (v1: Int, v2: Int) => v1 + v2,
      Seconds(60),
      Seconds(10)
    ).transform(a =>{
      val value = a.map(x =>(x._2,x._1))
      val value1 = value.sortByKey(false)
      val value2 = value1.map(x => (x._1,x._2))
      val tuples = value2.take(3)
      for(tup <- tuples){
        println(tup)
      }
      a
    }).print()
    sc.start()
    sc.awaitTermination()

  }
}
